package com.atguigu.flink.cep;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2023/2/3
 *
 *      套路:
 *              ①基于EventTime(有水印)
 *              ②定义一个Pattern
 *              ③把Pattern作用到 流上，得到一个模式流
 *              ④从模式流中获取匹配后的数据即可
 */
public class Demo1_HelloWorld
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());

        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        //定义模式
        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("匹配s1")
            .where(new SimpleCondition<WaterSensor>()
            {
                //如何过滤出想要的数据
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            });


        //把Pattern作用到 流上，得到一个模式流
        CEP.pattern(ds,pattern)
           .select(new PatternSelectFunction<WaterSensor, String>()
           {
               /*
                    从匹配到的数据中挑选感兴趣的，输出到下游

                    Map key是规则名
                        value: List<WaterSensor>> pattern  每来一条数据，应用规则匹配到的结果
                */
               @Override
               public String select(Map<String, List<WaterSensor>> pattern) throws Exception {
                   return pattern.toString();
               }
           })
           .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
